package org.apache.mina.filter.executor;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoFilterAdapter;
import org.apache.mina.common.IoFilterChain;
import org.apache.mina.common.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A filter that forward events to {@link Executor} in <a
 * href="http://dcl.mathcs.emory.edu/util/backport-util-concurrent/"
 * >backport-util-concurrent</a>. You can apply various thread model by
 * inserting this filter to the {@link IoFilterChain}.
 * 
 * 一个转发事件到Executor的过滤器。你可以通过让过滤器添加到IoFilterChain，让应用到各种各样的线程模型
 * <p>
 * Please note that this filter doesn't manage the life cycle of the underlying
 * {@link Executor}. You have to destroy or stop it by yourself.
 * 
 * 请注意，该filter不管理基本Executor的生命周期，你必须自己销毁或停止它。
 */
public class ExecutorFilter extends IoFilterAdapter {
    private final Logger   logger = LoggerFactory.getLogger(getClass());

    private final Executor executor;

    /**
     * Creates a new instance with the default thread pool implementation (
     * 
     * <tt>new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue() )</tt>
     * ).
     * 
     * 用一个默认的线程池实现创建新的实例。
     */
    public ExecutorFilter() {
        this(new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates a new instance with the specified <tt>executor</tt>.
     * 
     * 用指定的executor创建新的实例。
     */
    public ExecutorFilter(Executor executor) {
        if (executor == null) {
            throw new NullPointerException("executor");
        }

        this.executor = executor;
    }

    /**
     * Returns the underlying {@link Executor} instance this filter uses.
     * 
     * 返回该过滤器正在使用的期本Executor实例。
     */
    public Executor getExecutor() {
        return executor;
    }

    /**
     * 解决事件
     * 
     * @param nextFilter
     * @param session
     * @param type
     * @param data
     */
    private void fireEvent(NextFilter nextFilter, IoSession session,
            EventType type, Object data) {
        Event event = new Event(type, nextFilter, data);
        SessionBuffer buf = SessionBuffer.getSessionBuffer(session);

        boolean execute;
        synchronized (buf.eventQueue) {
            buf.eventQueue.offer(event);
            if (buf.processingCompleted) {
                buf.processingCompleted = false;
                execute = true;
            } else {
                execute = false;
            }
        }

        if (execute) {
            if (logger.isDebugEnabled()) {
                logger.debug("Launching thread for "
                        + session.getRemoteAddress());
            }

            executor.execute(new ProcessEventsRunnable(buf));
        }
    }

    /**
     * 
     */
    private static class SessionBuffer {
        private static final String KEY = SessionBuffer.class.getName()
                                                + ".KEY";

        private static SessionBuffer getSessionBuffer(IoSession session) {
            synchronized (session) {
                SessionBuffer buf = (SessionBuffer) session.getAttribute(KEY);
                if (buf == null) {
                    buf = new SessionBuffer(session);
                    session.setAttribute(KEY, buf);
                }
                return buf;
            }
        }

        private final IoSession    session;

        private final Queue<Event> eventQueue          = new LinkedList<Event>();

        private boolean            processingCompleted = true;

        private SessionBuffer(IoSession session) {
            this.session = session;
        }
    }

    /**
     * 事件类型
     */
    protected static class EventType {
        public static final EventType OPENED    = new EventType("OPENED");

        public static final EventType CLOSED    = new EventType("CLOSED");

        public static final EventType READ      = new EventType("READ");

        public static final EventType WRITTEN   = new EventType("WRITTEN");

        public static final EventType RECEIVED  = new EventType("RECEIVED");

        public static final EventType SENT      = new EventType("SENT");

        public static final EventType IDLE      = new EventType("IDLE");

        public static final EventType EXCEPTION = new EventType("EXCEPTION");

        private final String          value;

        private EventType(String value) {
            this.value = value;
        }

        public String toString() {
            return value;
        }
    }

    /**
     * 事件
     */
    protected static class Event {
        // 事件类型
        private final EventType  type;

        // 下一个过滤器
        private final NextFilter nextFilter;

        // 数据
        private final Object     data;

        Event(EventType type, NextFilter nextFilter, Object data) {
            this.type = type;
            this.nextFilter = nextFilter;
            this.data = data;
        }

        public Object getData() {
            return data;
        }

        public NextFilter getNextFilter() {
            return nextFilter;
        }

        public EventType getType() {
            return type;
        }
    }

    public void sessionCreated(NextFilter nextFilter, IoSession session) {
        nextFilter.sessionCreated(session);
    }

    public void sessionOpened(NextFilter nextFilter, IoSession session) {
        fireEvent(nextFilter, session, EventType.OPENED, null);
    }

    public void sessionClosed(NextFilter nextFilter, IoSession session) {
        fireEvent(nextFilter, session, EventType.CLOSED, null);
    }

    public void sessionIdle(NextFilter nextFilter, IoSession session,
            IdleStatus status) {
        fireEvent(nextFilter, session, EventType.IDLE, status);
    }

    public void exceptionCaught(NextFilter nextFilter, IoSession session,
            Throwable cause) {
        fireEvent(nextFilter, session, EventType.EXCEPTION, cause);
    }

    public void messageReceived(NextFilter nextFilter, IoSession session,
            Object message) {
        fireEvent(nextFilter, session, EventType.RECEIVED, message);
    }

    public void messageSent(NextFilter nextFilter, IoSession session,
            Object message) {
        fireEvent(nextFilter, session, EventType.SENT, message);
    }

    protected void processEvent(NextFilter nextFilter, IoSession session,
            EventType type, Object data) {
        if (type == EventType.RECEIVED) {
            nextFilter.messageReceived(session, data);
        } else if (type == EventType.SENT) {
            nextFilter.messageSent(session, data);
        } else if (type == EventType.EXCEPTION) {
            nextFilter.exceptionCaught(session, (Throwable) data);
        } else if (type == EventType.IDLE) {
            nextFilter.sessionIdle(session, (IdleStatus) data);
        } else if (type == EventType.OPENED) {
            nextFilter.sessionOpened(session);
        } else if (type == EventType.CLOSED) {
            nextFilter.sessionClosed(session);
        }
    }

    public void filterWrite(NextFilter nextFilter, IoSession session,
            WriteRequest writeRequest) {
        nextFilter.filterWrite(session, writeRequest);
    }

    public void filterClose(NextFilter nextFilter, IoSession session)
            throws Exception {
        nextFilter.filterClose(session);
    }

    private class ProcessEventsRunnable implements Runnable {
        private final SessionBuffer buffer;

        ProcessEventsRunnable(SessionBuffer buffer) {
            this.buffer = buffer;
        }

        public void run() {
            while (true) {
                Event event;

                synchronized (buffer.eventQueue) {
                    event = buffer.eventQueue.poll();

                    if (event == null) {
                        buffer.processingCompleted = true;
                        break;
                    }
                }

                processEvent(event.getNextFilter(), buffer.session, event
                        .getType(), event.getData());
            }

            if (logger.isDebugEnabled()) {
                logger.debug("Exiting since queue is empty for "
                        + buffer.session.getRemoteAddress());
            }
        }
    }
}
